"use strict";

var zookeeper = require('node-zookeeper-client');
var amqp = require('amqplib/callback_api');
var util = require('util');
var EventEmitter = require('events').EventEmitter;

// zk
var __zkNode = '/myrpc/server/currid';
var __zkConfig = 'localhost:2181';
// var __zkConfig = '192.168.102.142:2181';
var __zkBaseId = '1';

// mq
var __rabbitmqConfig = 'amqp://localhost';
// var __rabbitmqConfig = 'amqp://192.168.102.142';
var __qRequest = 'request_to_{type}';
var __qReply = 'reply_to_{appId}';


/**
 * 
 */
function RpcFrame() {
	this.__status = 'before_init';
}
util.inherits(RpcFrame, EventEmitter);
module.exports = RpcFrame;

/**
 * create a server and run it
 */
RpcFrame.prototype.createServer = function(server, cb) {
	this.createServerOrClient(true, server, cb);
}

/**
 * create a server and run it
 */
RpcFrame.prototype.createClient = function(server, cb) {
	this.createServerOrClient(false, server, cb);
}

RpcFrame.prototype.createServerOrClient = function(isServer, server, cb) {
	if (typeof server === 'function') {
		server = new server();
	}

	var self = this;
	self.__server = server;

	self.on('done_mq', function() {
		self.__status = 'done_mq';
		self.prepareZk();
	});
	self.on('done_zk', function() {
		self.__status = 'done_zk';
		self.getNextId();
	});

	self.on('done_id', function(myid) {
		self.__status = 'done_id';
		self.__id = myid;
		if (isServer) {
			self.prepareServer(myid, server);
		} else {
			self.prepareClient(myid, server);
		}
	});

	self.on('done_server', function() {
		self.__status = 'done_server';
		self.done_server = true;
		cb && cb();
	});

	self.on('done_client', function() {
		self.__status = 'done_client';
		self.done_client = true;
		cb && cb();
	});

	// start here
	if (self.__status === 'before_init') {
		self.prepareMq();
	} else {
		if (isServer) {
			self.prepareServer(self.__id, server);
		} else {
			self.prepareClient(self.__id, server);
		}
	}
}



/** get id from zookeeper */
RpcFrame.prototype.getNextId = function() {
	console.log('getNextId');
	var self = this;
	if (self.__id !== undefined) {
		return self.emit('getId', self.__id);
	}
	var zk = self.__zk;

	// if exists read and save
	// not exist mkdir and save

	function getId() {
		zk.getData(__zkNode, null, function(err, data, stat) {
			// console.log('getData node %s', __zkNode);
			if (err) return self.emit('error', err);
			var nextId = parseInt(data) + 1;
			zk.setData(__zkNode, new Buffer(nextId + ''), -1, function(err, stat) {
				if (err) return self.emit('error', err);
				return self.emit('done_id', nextId);
			});
		});
	}

	function firstSetId() {
		zk.mkdirp(__zkNode, new Buffer(__zkBaseId + ''), function(err, path) {
			// console.log('create node %s', __zkNode);
			if (err) return self.emit('error', err);
			self.emit('done_id', parseInt(__zkBaseId));
		});
	}


	zk.exists(__zkNode, function(err, stat) {
		if (err) return self.emit('error', err);
		if (stat) { // if exists read and save
			// console.log('exists node %s', __zkNode);
			getId();
		} else { // not exist mkdir and save
			// console.log('not exists node %s', __zkNode);
			firstSetId();
		}
	});
}

/**
 *
 */
RpcFrame.prototype.prepareServer = function(myid) {
	console.log('prepareServer');
	var self = this;
	var conn = self.__mqConn;
	if (self.__mqch) {
		prepareMqRequestChannel();
		return;
	}
	conn.createChannel(function(err, ch) {
		if (err) return self.emit('error', err);
		self.__mqCh = ch;
		prepareMqRequestChannel();
	});

	function prepareMqRequestChannel() {
		var ch = self.__mqCh;
		var qRequst = __qRequest.replace('{type}', self.__server.type);
		console.log('create queue %s', qRequst);
		ch.assertQueue(qRequst, {
			durable: true
		});
		ch.prefetch(1); // TODO fetch reply should not set prefetch =1 ,but fetch request should set to 1 
		ch.consume(qRequst, function(msg) {
			console.log('receive request at ', new Date().getTime());
			console.log('receive request:%s', msg.content);
			ch.ack(msg);
			handleMsg(msg);
		}, {
			noAck: false
		});
		self.prepared = true;
		self.emit('done_server');
	}

	function handleMsg(msg) { // msg: {route:"xx.xx.xx",args:[xx,xx]}
		console.log('handleMsg');
		// self.__server;
		var result = 'TODO';
		reply(msg, result);
	}

	function reply(msg, result) {
		console.log('send reply =%s', result);
		var ch = self.__mqCh;
		var replyTo = msg.properties.replyTo;
		var replyQid = __qReply.replace('{appId}', replyTo);
		console.log('send reply at ', new Date().getTime());
		ch.sendToQueue(replyQid, new Buffer(result), {
			correlationId: msg.properties.correlationId
		});
	}
}

RpcFrame.prototype.prepareClient = function(myid) {
	console.log('prepareClient');
	var self = this;
	var conn = self.__mqConn;
	self.rpcCallback = {};
	if (self.__mqch) {
		prepareMqReplyChannel();
		return;
	}
	conn.createChannel(function(err, ch) {
		if (err) return self.emit('error', err);
		self.__mqCh = ch;
		prepareMqReplyChannel();
	});

	function prepareMqReplyChannel() {
		var ch = self.__mqCh;
		var qReply = __qReply.replace('{appId}', self.appId);
		ch.assertQueue(qReply, {
			exclusive: true
		}, function(err, q) {
			if (err) return self.emit('error', err);
			console.log('create queue %s', qReply);
			ch.consume(qReply, function(msg) {
				console.log('receive reply at ', new Date().getTime());
				ch.ack(msg); // ack
				console.log('receive reply: %s', msg.content);
				var msgId = msg.properties.correlationId;
				var cb = self.rpcCallback[msgId];
				if (!cb) {
					console.error('no callback with msgId = %s', msgId);
				} else {
					cb(err, msg.content);
				}
			});
			self.emit('done_client');
		});
	}
}

/**
 * cb( err, function(result){...})
 */
RpcFrame.prototype.rpc = function(path, arg, cb) {
	var self = this;
	if (!self.done_client) return self.emit('error', new Error('client not ready'));

	// send to mq
	var ch = self.__mqCh;
	var qSend = __qRequest.replace('{type}', self.__server.type);
	ch.assertQueue(qSend, {
		durable: true
	}, function(err, q) {
		if (!self.msgId) self.msgId = 1;
		var msgId = (self.msgId++) + '';
		console.log('send request at ', new Date().getTime());
		ch.sendToQueue(qSend, new Buffer(arg), {
			correlationId: msgId,
			replyTo: self.appId + '',
			persisitent: true,
		});
		// put callback to list
		self.rpcCallback[msgId] = cb;
	});
};


RpcFrame.prototype.prepareZk = function() {
	console.log('prepareZk');
	var self = this;
	var zk = zookeeper.createClient(__zkConfig);
	zk.connect();
	zk.once('connected', function() {
		console.log('zk connected');
		self.__zk = zk;
		self.emit('done_zk');
	});
};


RpcFrame.prototype.prepareMq = function() {
	console.log('prepareMq');
	var self = this;
	amqp.connect(__rabbitmqConfig, function(err, conn) {
		if (err) return self.emit('error', err);
		console.log('mq connected');
		self.__mqConn = conn;
		self.emit('done_mq');
	});
};